package com.atguigu.tingshu.search.config;

import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    /**
     * 申明topic的配置
     * @return
     */
    @Bean
    public NewTopic newTopic(){
        return TopicBuilder
                .name("album_stat_topic")//指定topic的名字
                // .name("tingshu_stream_rank_topic")//指定topic的名字
                .partitions(4)//分区的数量: 下标0开始
                .replicas(3) // 集群的实例数量
                .build();
    }

    /**
     * 流式编程的配置
     * @return
     */
    @Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kafkaStreamsConfiguration(){
        //配置的初始化
        Map<String, Object> configs = new HashMap<>();
        //服务的ip地址和端口号
        configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        //应用的别名
        configs.put(StreamsConfig.APPLICATION_ID_CONFIG, "aid-service-search");
        //流式消费者的别名
        configs.put(StreamsConfig.CLIENT_ID_CONFIG, "cid-service-search");
        configs.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        configs.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        //初始化流式编程的配置对象
        KafkaStreamsConfiguration configuration =
                new KafkaStreamsConfiguration(configs);
        return configuration;
    }
}
